package yuan.redislock.blocking_queue;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import org.springframework.data.redis.core.RedisTemplate;

import java.lang.reflect.Type;
import java.util.Set;
import java.util.UUID;

public class RedisDelayingQueue<T> {
    static class TaskItem<T>{
        public String id;
        public T msg;
    }
    private Type TaskType = new TypeReference<TaskItem<T>>() { }.getType();
    private RedisTemplate<String,String> redisTemplate;
    private String queueKey;

    public RedisDelayingQueue(RedisTemplate<String,String> redisTemplate, String queueKey) {
        this.redisTemplate = redisTemplate;
        this.queueKey = queueKey;
    }

    public void delay(T msg) {
        TaskItem task = new TaskItem();
        task.id = UUID.randomUUID().toString();
        task.msg = msg;
        String s = JSON.toJSONString(task);
        //加入延时队列，5s后再试
        redisTemplate.opsForZSet().add(queueKey,s, System.currentTimeMillis() + 5000);

    }

    public void loop(){
        while (!Thread.interrupted()) {
            Set<String> vals = redisTemplate.opsForZSet().rangeByScore(queueKey, 0, System.currentTimeMillis(), 0, 1);
            if (vals.isEmpty()) {
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    break;
                }
                continue;
            }
            String s = String.valueOf(vals.iterator().next());
            //多线程并发争抢成功的关键
            if (redisTemplate.opsForZSet().remove(queueKey, s) > 0) {
                TaskItem parse = JSON.parseObject(s, TaskType);
                this.handleMsg(parse);
            }
        }



    }

    public void handleMsg(TaskItem taskItem) {
        System.out.println(taskItem.msg+" hand");
    }



}
